/*
 * MIT License
 *
 * Copyright (c) 2024 org.zzz (https://gitee.com/frostforest) (https://gitee.com/frostforest)
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

package org.zzz.pcdc;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.zzz.pcdc.PcdcUtils.getFirst;

/**
 * Pcdc（Polling Change Data Capture）构建器类，用于配置和创建Pcdc实例。
 * <p>
 * 该构建器采用流式API设计，允许用户通过链式调用方法来配置Pcdc实例的各种参数，
 * 包括表配置、线程配置、数据处理回调等。
 * <p>
 * 使用示例:
 * <pre>
 * Pcdc pcdc = new PcdcBuilder(dataSource, "orders")
 *     .idColumn("id")
 *     .updatedAtColumn("updated_at")
 *     .dataHandler((tableConfig, records) -> {
 *         // 处理数据变更
 *     })
 *     .build();
 * </pre>
 *
 * @since 1.0
 */
public class PcdcBuilder {
    private final DataSource dataSource;
    private final String tableName;
    private String idColumnName = "id";
    private String updatedAtColumnName = "updated_at";
    private String[] columnsToReport = null;

    // 线程配置参数
    private int pollingIntervalSeconds = PcdcThreadConfig.DEFAULT_POLLING_INTERVAL_SECONDS;
    private int throttleSeconds = PcdcThreadConfig.DEFAULT_THROTTLE_SECONDS;
    private int batchWorkSize = PcdcThreadConfig.DEFAULT_BATCH_WORK_SIZE;
    private int batchWorkTimeoutSeconds = PcdcThreadConfig.DEFAULT_BATCH_WORK_TIMEOUT_SECONDS;
    private int partitionThreshold = PcdcThreadConfig.DEFAULT_PARTITION_THRESHOLD;
    private int minPartitionIntervalSeconds = PcdcThreadConfig.DEFAULT_MIN_PARTITION_INTERVAL_SECONDS;
    private int workerPoolSize = PcdcThreadConfig.DEFAULT_WORKER_POOL_SIZE;

    private PcdcDataHandler dataHandler;
    private PcdcErrorHandler errorHandler;

    /**
     * 创建Pcdc构建器实例。
     * <p>
     * 此构造函数初始化构建器，设置必要的数据源和要监控的表名。
     * 其他配置参数将使用默认值，可以通过链式调用其他方法进行自定义。
     *
     * @param dataSource 数据库连接池，用于获取数据库连接
     * @param tableName  要监控变更的MySQL表名
     * @throws NullPointerException 如果dataSource或tableName为null
     * @since 1.0
     */
    public PcdcBuilder(DataSource dataSource, String tableName) {
        this.dataSource = dataSource;
        this.tableName = tableName;
    }

    /**
     * 设置数据处理回调函数。
     * <p>
     * 数据处理回调是Pcdc的核心组件，用于处理检测到的数据变更。
     * 当Pcdc检测到表中的数据变更时，会调用此回调函数，并传入表配置和变更的记录列表。
     * <p>
     * 此回调必须设置，否则在调用build()方法时会抛出异常。
     *
     * @param dataHandler 数据处理回调函数，接收表配置和变更记录列表作为参数
     * @return 当前构建器实例，用于链式调用
     * @throws NullPointerException 如果dataHandler为null
     * @since 1.0
     */
    public PcdcBuilder dataHandler(PcdcDataHandler dataHandler) {
        this.dataHandler = dataHandler;
        return this;
    }

    /**
     * 设置错误处理回调函数。
     * <p>
     * 错误处理回调用于处理Pcdc在运行过程中遇到的异常情况。
     * 当Pcdc在轮询、处理数据变更或其他操作中遇到异常时，会调用此回调函数。
     * <p>
     * 如果不设置错误处理回调，Pcdc将使用默认的错误处理策略，通常是记录错误并继续运行。
     *
     * @param errorHandler 错误处理回调函数，接收异常对象作为参数
     * @return 当前构建器实例，用于链式调用
     * @since 1.0
     */
    public PcdcBuilder errorHandler(PcdcErrorHandler errorHandler) {
        this.errorHandler = errorHandler;
        return this;
    }

    /**
     * 设置主键列名，只支持数值型主键。
     * <p>
     * 此列应该是表的主键或具有唯一约束的列，用于跟踪记录的变更。
     * 如果不设置，将使用默认值"id"。
     *
     * @param idColumnName ID列名，必须是数值类型
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果列名不符合MySQL标识符规则或是保留关键字
     * @since 1.0
     */
    public PcdcBuilder idColumn(String idColumnName) {
        this.idColumnName = idColumnName;
        return this;
    }

    /**
     * 设置更新时间戳列名。
     * <p>
     * 此列必须是 datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP 类型，
     * 用于确定哪些记录发生了变更。如果不设置，将使用默认值"updated_at"。
     * <p>
     * 重要说明：
     * <ul>
     *   <li>必须绝对避免在应用层设置此列的值，应完全由数据库维护</li>
     *   <li>否则不保证数据变更处理的完整性</li>
     *   <li>如果需要绝对保证数据变更处理的完整性并且能接受一定的性能损耗，可以考虑使用触发器</li>
     * </ul>
     *
     * @param updatedAtColumnName 更新时间戳列名，秒级精度，超出忽略
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果列名不符合MySQL标识符规则或是保留关键字
     * @since 1.0
     */
    public PcdcBuilder updatedAtColumn(String updatedAtColumnName) {
        this.updatedAtColumnName = updatedAtColumnName;
        return this;
    }

    /**
     * 设置要报告的列。
     * <p>
     * 此方法允许指定在数据变更时要包含在结果中的列。
     * 如果不设置，默认会获取所有字段（相当于SELECT *）。
     * <p>
     * 列名可以是逗号分隔的字符串，例如："id, name, age"。
     * 列名会自动进行规范化处理，添加反引号以避免与MySQL关键字冲突。
     *
     * @param columnsToReport 数据变更时要报告的列，逗号分隔的字符串
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果任何列名不符合MySQL标识符规则或是保留关键字
     * @since 1.0
     */
    public PcdcBuilder columnsToReport(String columnsToReport) {
        if (columnsToReport != null && !columnsToReport.trim().isEmpty()) {
            this.columnsToReport = columnsToReport.split("\\s*,\\s*");
        }
        return this;
    }

    /**
     * 设置要报告的列（可变参数版本）。
     * <p>
     * 此方法允许通过可变参数指定在数据变更时要包含在结果中的列。
     * 如果不设置，默认会获取所有字段（相当于SELECT *）。
     * <p>
     * 列名会自动进行规范化处理，添加反引号以避免与MySQL关键字冲突。
     *
     * @param columnsToReport 数据变更时要报告的列，可变参数形式
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果任何列名不符合MySQL标识符规则或是保留关键字
     * @since 1.0
     */
    public PcdcBuilder columnsToReport(String... columnsToReport) {
        this.columnsToReport = columnsToReport;
        return this;
    }

    /**
     * 设置变更数据批处理大小。
     * <p>
     * 批处理大小决定了每次处理多少条变更记录。较大的批处理大小可以提高处理效率，
     * 但会增加内存使用和单次处理的复杂度；较小的批处理大小会降低处理效率，但更容易处理。
     * <p>
     * 如果不设置，将使用默认值（通常为100条记录）。
     *
     * @param batchWorkSize 批处理大小，必须大于0
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果批处理大小小于或等于0
     * @since 1.0
     */
    public PcdcBuilder batchWorkSize(int batchWorkSize) {
        this.batchWorkSize = batchWorkSize;
        return this;
    }


    /**
     * 设置分区阈值。
     * <p>
     * 分区阈值用于控制数据处理的分区策略。每次系统重启时，
     * 会检查积压的数据量，如果超过该阈值，系统将启用分区处理模式，
     * 将大量数据分成多个小批次进行处理，以提高处理效率和减少资源消耗。
     * <p>
     * 如果不设置，将使用默认值（通常为1000条记录）。
     *
     * @param partitionThreshold 分区阈值，必须大于0
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果分区阈值小于或等于0
     * @since 1.0
     */
    public PcdcBuilder partitionThreshold(int partitionThreshold) {
        this.partitionThreshold = partitionThreshold;
        return this;
    }

    /**
     * 设置分区最小时间跨度。
     * <p>
     * 分区的最小时间跨度，优先级高于分区阈值。
     * <p>
     *
     * @param minPartitionIntervalSeconds 分区最小时间跨度，必须大于0
     * @return 当前构建器实例，用于链式调用
     * @since 1.0
     */
    public PcdcBuilder minPartitionIntervalSeconds(int minPartitionIntervalSeconds) {
        this.minPartitionIntervalSeconds = minPartitionIntervalSeconds;
        return this;
    }

    /**
     * 设置轮询间隔秒数。
     * <p>
     * 轮询间隔决定了系统检查数据变更的频率。较短的轮询间隔可以更快地检测到变更，
     * 但会增加数据库负载；较长的轮询间隔会减少数据库负载，但会增加变更检测的延迟。
     * <p>
     * 如果不设置，将使用默认值（通常为5秒）。
     * <p>
     * 注意：在高负载系统中，建议适当增加轮询间隔，以减少对数据库的压力。
     *
     * @param pollingIntervalSeconds 轮询间隔秒数，必须大于0
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果轮询间隔小于或等于0
     * @since 1.0
     */
    public PcdcBuilder pollingIntervalSeconds(int pollingIntervalSeconds) {
        this.pollingIntervalSeconds = pollingIntervalSeconds;
        return this;
    }

    /**
     * 设置节流秒数。
     * <p>
     * 节流机制用于控制数据处理的速率，防止系统过载。系统不会单独处理每一条数据变更，
     * 而是将一个时间范围内的变更聚合处理。throttleSeconds定义了这个时间范围的最小值。
     * <p>
     * 注意：throttleSeconds的优先级高于{@link #partitionThreshold(int)}。
     * 在决定何时处理数据变更时，系统会优先考虑节流时间设置。
     * <p>
     * 如果不设置，将使用默认值（通常为0秒，表示不启用节流）。
     *
     * @param throttleSeconds 节流秒数，必须大于或等于0
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果节流秒数小于0
     * @since 1.0
     */
    public PcdcBuilder throttleSeconds(int throttleSeconds) {
        this.throttleSeconds = throttleSeconds;
        return this;
    }

    /**
     * 设置批处理超时秒数。
     * <p>
     * 批处理超时用于防止并发冲突和处理卡死。系统使用数据库实现分布式锁控制，
     * 确保同一批数据不会被多个实例同时处理。如果一个批处理操作超过指定的超时时间，
     * 系统会认为该操作已失败或卡死，并允许其他实例接管处理。
     * <p>
     * 在分布式环境中，合理设置此值对于系统的稳定性和可靠性至关重要。
     * <p>
     * 如果不设置，将使用默认值（通常为60秒）。
     *
     * @param batchWorkTimeoutSeconds 批处理超时秒数，必须大于0
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果批处理超时秒数小于或等于0
     * @since 1.0
     */
    public PcdcBuilder batchWorkTimeoutSeconds(int batchWorkTimeoutSeconds) {
        this.batchWorkTimeoutSeconds = batchWorkTimeoutSeconds;
        return this;
    }

    /**
     * 设置工作线程池大小。
     * <p>
     * 工作线程池用于并行处理数据变更。较大的线程池可以提高处理效率，
     * 但会消耗更多的系统资源；较小的线程池会减少资源消耗，但可能降低处理效率。
     * <p>
     * 建议根据系统的CPU核心数和内存情况来设置合适的线程池大小。
     * 在高负载系统中，过大的线程池可能导致资源竞争和性能下降。
     * <p>
     * 如果不设置，将使用默认值（通常为可用处理器数量）。
     *
     * @param workerPoolSize 工作线程池大小，必须大于0
     * @return 当前构建器实例，用于链式调用
     * @throws IllegalArgumentException 如果工作线程池大小小于或等于0
     * @since 1.0
     */
    public PcdcBuilder workerPoolSize(int workerPoolSize) {
        this.workerPoolSize = workerPoolSize;
        return this;
    }

    /**
     * 构建Pcdc实例。
     * <p>
     * 此方法根据之前设置的所有配置参数创建并初始化一个新的Pcdc实例。
     * 在构建过程中，会验证所有必需的配置参数，并应用默认值到未设置的可选参数。
     * <p>
     * 构建过程包括以下步骤：
     * <ol>
     *   <li>验证必需的配置参数（如数据源、表名、数据处理回调等）</li>
     *   <li>创建表配置和线程配置对象</li>
     *   <li>检查进度表是否存在</li>
     *   <li>验证必要的数据库索引是否存在</li>
     *   <li>创建并返回配置完成的Pcdc实例</li>
     * </ol>
     * <p>
     * 注意：构建完成后，所有配置参数将被锁定，无法再进行修改。
     * 如需不同的配置，请创建新的PcdcBuilder实例。
     *
     * @return 配置完成的Pcdc实例
     * @throws IllegalStateException    如果缺少必需的配置参数
     * @throws SQLException             如果在访问数据库时发生错误
     * @throws IllegalArgumentException 如果表缺少必要的索引
     * @throws RuntimeException         如果进度表不存在
     * @since 1.0
     */
    public Pcdc build() throws SQLException {
        validateRequiredParameters();

        // 创建不可变的配置对象
        PcdcTableConfig tableConfig = new PcdcTableConfig(
                tableName,
                updatedAtColumnName,
                idColumnName,
                columnsToReport
        );

        PcdcThreadConfig threadConfig = new PcdcThreadConfig(
                pollingIntervalSeconds,
                throttleSeconds,
                batchWorkSize,
                batchWorkTimeoutSeconds,
                partitionThreshold,
                minPartitionIntervalSeconds,
                workerPoolSize
        );

        String progressTableName = getProgressTableName();
        if (!checkProgressTable()) {
            throw new RuntimeException("Table not found: " + progressTableName);
        }
        PcdcIndexMapBuilder indexFinder = new PcdcIndexMapBuilder(dataSource);
        Map<String, List<String>> canUseIndexMapInCdcTable = indexFinder.buildCanUseIndexMap(tableName);
        String updatedAtDotIdIndex = getUpdatedAtDotIdIndex(canUseIndexMapInCdcTable);
        if (updatedAtDotIdIndex == null) {
            throw new IllegalArgumentException("Table " + tableName + " must have an index (updated_at asc, id asc)");
        }
        Map<String, List<String>> progressTableIndex = indexFinder.buildCanUseIndexMap(progressTableName);
        List<String> indexIsComplete = progressTableIndex.get(Pcdc.INDEX_IS_COMPLETE_INDEX);
        if (indexIsComplete == null || indexIsComplete.isEmpty()) {
            throw new IllegalArgumentException("Table %s must have an index (%s)"
                    .formatted(progressTableName, Pcdc.INDEX_IS_COMPLETE_INDEX));
        }
        String isCompleteDotlockExpireTimeDotUpperBoundaryIndex = getFirst(indexIsComplete);
        return new Pcdc(dataSource,
                tableConfig,
                threadConfig,
                dataHandler,
                errorHandler,
                progressTableName,
                progressTableIndex,
                isCompleteDotlockExpireTimeDotUpperBoundaryIndex,
                canUseIndexMapInCdcTable,
                updatedAtDotIdIndex);
    }

    private String getUpdatedAtDotIdIndex(Map<String, List<String>> canUseIndexMapInCdcTable) {
        String updatedAtDotId = Stream.of(updatedAtColumnName,
                        idColumnName)
                .map(String::strip)
                .map(String::toLowerCase)
                .collect(Collectors.joining(","));
        List<String> inddexList = canUseIndexMapInCdcTable.getOrDefault(updatedAtDotId, Collections.EMPTY_LIST);
        return getFirst(inddexList);
    }

    private boolean checkProgressTable() throws SQLException {
        try (Connection conn = dataSource.getConnection()) {
            DatabaseMetaData metaData = conn.getMetaData();
            String[] types = {"TABLE"}; // 只检查表，忽略视图等

            try (ResultSet rs = metaData.getTables(null,
                    null,
                    getProgressTableName(),
                    types)) {
                return rs.next(); // 如果结果集有下一行，表示表存在
            }
        }
    }

    private String getProgressTableName() {
        return PcdcProgress.TABLE_NAME_PREFIX + tableName;
    }

    /**
     * 验证必要参数
     *
     * @throws IllegalStateException 如果缺少必要参数
     */
    private void validateRequiredParameters() {
        if (dataSource == null) {
            throw new IllegalStateException("数据源不能为空");
        }
        if (tableName == null || tableName.trim().isEmpty()) {
            throw new IllegalStateException("表名不能为空");
        }
        if (dataHandler == null) {
            throw new IllegalStateException("数据处理回调不能为空");
        }
    }
}
